package com.leigq.www.producer.service;

import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.leigq.www.common.constant.QueueConstant;
import com.leigq.www.common.entity.Order;
import com.leigq.www.common.enumeration.QueueEnum;
import com.leigq.www.producer.entity.BrokerMessageLog;
import com.leigq.www.producer.enumeration.MessageLogStatusEnum;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.util.Objects;

/**
 * RabbitMQ订单发送
 * <br/>
 * 参考：<a href='https://blog.csdn.net/wangpf2011/article/details/86600809'>分布式架构核心组件之消息队列RabbitMQ</a>
 * <p>
 * 创建人：asus <br>
 * 创建时间：2019-02-16 15:04 <br>
 * <p>
 * 修改人： <br>
 * 修改时间： <br>
 * 修改备注： <br>
 * </p>
 */
@Component
@Slf4j
public class OrderSender {

    /**
     * The Rabbit template.
     */
    @Autowired
    private RabbitTemplate rabbitTemplate;

    /**
     * The Message log service.
     */
    @Autowired
    private BrokerMessageLogService messageLogService;

    /**
     * The Object mapper.
     */
    @Autowired
    private ObjectMapper objectMapper;

    /**
     * 回调函数(只能设置一个): confirm 消息确认
     * <br>创建人： asus
     * <br>创建时间： 2019-02-18 17:17
     * <br>
     */
    private final RabbitTemplate.ConfirmCallback confirmCallback = (correlationData, ack, cause) -> {
        log.info("correlationData = [{}]", correlationData);
        log.info("cause = [{}]", cause);
        if (Objects.isNull(correlationData)) {
            log.error("消息发送异常....1");
            return;
        }
        String messageId = correlationData.getId();

        if (ack) {
            // 消息发送成功
            // 更新消息状态为成功：1
            BrokerMessageLog log = new BrokerMessageLog();
            log.setStatus(MessageLogStatusEnum.SUCCESS.getValue());
            log.setMsgId(messageId);
            messageLogService.update(log);
        } else {
            // 失败则进行具体的后续操作:重试 或者补偿等手段
            log.error("消息发送异常....2");
        }
    };

    /**
     * 如果出现路由键不可达的情况，则使用监听器对不可达的消息进行后续处理回调函数(只能设置一个):
     * <br>创建人： asus
     * <br>创建时间： 2019-02-18 17:17
     * <br>
     */
    private final RabbitTemplate.ReturnCallback returnCallBack = (message, replyCode, replyText, exchange, routingKey) -> {
        log.warn("返回消息回调:{} 应答代码:{} 回复文本:{} 交换器:{} 路由键:{}", message, replyCode, replyText, exchange, routingKey);
    };

    /**
     * 发送消息方法调用: 构建自定义对象消息
     *
     * @param order the order
     */
    public void sendOrder(Order order) {

        // 通过实现 ConfirmCallback 接口，消息发送到 Broker 后触发回调，确认消息是否到达 Broker 服务器，也就是只确认是否正确到达 Exchange 中
        rabbitTemplate.setConfirmCallback(confirmCallback);

        // 通保证消息对 broker 端是可达的，如果出现路由键不可达的情况，则使用监听器对不可达的消息进行后续处理，保证消息的路由成功
        rabbitTemplate.setReturnCallback(returnCallBack);

        // 设置消息唯一ID，这里从 订单 中获取消息唯一ID
        CorrelationData correlationData = new CorrelationData(order.getMsgId());
        try {
            rabbitTemplate.convertAndSend(
                    QueueConstant.ORDER_QUEUE_EXCHANGE_NAME,
                    QueueConstant.ORDER_QUEUE_ROUTING_KEY,
                    objectMapper.writeValueAsString(order),
                    (msg) -> {
                        final MessageProperties messageProperties = msg.getMessageProperties();
                        messageProperties.setHeader("x-retries", 0);
                        return msg;
                    },
                    correlationData
            );
        } catch (JsonProcessingException e) {
            log.error("json转化异常:", e);
        }
    }

    /**
     * 发送消息测试延迟队列
     * <br/>
     * 方式一
     *
     * @param order       发送对象
     * @throws JsonProcessingException the json processing exception
     */
    public void sendDelay1(Order order, int millisecond) throws JsonProcessingException {
        this.rabbitTemplate.convertAndSend(
                QueueEnum.NORMAL_QUEUE1.getExchangeName(),
                QueueEnum.NORMAL_QUEUE1.getRoutingKey(),
                objectMapper.writeValueAsString(order),
                (msg) -> {
                    final MessageProperties messageProperties = msg.getMessageProperties();
                    // 可以在发消息时，针对每条消息设置TTL，如果同时也设置了队列的TTL属性， 以最先到期的时间为准
                    messageProperties.setExpiration(millisecond + "");
                    return msg;
                }
        );
        log.info("发送消息至 Queue: [{}] 队列成功 ：order = [{}]", QueueEnum.NORMAL_QUEUE1.getQueueName(), order);
    }


    /**
     * 发送消息测试延迟队列
     * <br/>
     * 方式二
     *
     * @param order       发送对象
     * @param millisecond 延时（毫秒）
     * @throws JsonProcessingException the json processing exception
     */
    public void sendDelay2(Order order, int millisecond) throws JsonProcessingException {
        this.rabbitTemplate.convertAndSend(
                QueueEnum.DELAY_QUEUE2.getExchangeName(),
                QueueEnum.DELAY_QUEUE2.getRoutingKey(),
                objectMapper.writeValueAsString(order),
                (msg) -> {
                    final MessageProperties messageProperties = msg.getMessageProperties();
                    // 可以在发消息时，针对每条消息设置延迟时间，和上面TTL不一样，Delay属性是针对于x-delay-message的
                    messageProperties.setDelay(millisecond);
                    return msg;
                }
        );
        log.info("发送消息至 Queue: [{}] 延迟队列成功 ：order = [{}]", QueueEnum.DELAY_QUEUE2.getQueueName(), order);
    }
}
